﻿

#include "AsioSocketAsync.hpp"
#include "cybertron/network/SocketEventHandler.hpp"
#include "cybertron/core/Log.hpp"
#include "cybertron/core/UtilTime.hpp"
#include "cybertron/network/perf.hpp"

CYBERTRON_BEGIN

/*

	错误码（十进制）
	枚举
	发现原因

	10009
	boost::asio::error::bad_descriptor
	在一个已经关闭了的套接字上执行async_receive()

	995
	boost::asio::error::operation_aborted
	正在async_receive()异步任务等待时，本端关闭套接字

	10054
	boost::asio::error::connection_reset
	正在async_receive()异步任务等待时，远端的TCP协议层发送RESET终止链接，暴力关闭套接字。常常发生于远端进程强制关闭时，操作系统释放套接字资源。

	2
	boost::asio::error::eof
	正在async_receive()异步任务等待时，远端关闭套接字，这里跟10054发生的情况似乎一样，但是实际上应该是有区别的，这个是远端正常关闭套接字。

*/

AsioSocketAsync::AsioSocketAsync(
	std::shared_ptr<asio::io_context> pContext,
	asio::ip::tcp::socket&& socket,
	ESocketStatus status,
	const std::string remoteIp,
	int remotePort) : 
	mSocket(std::move(socket))
{
	mLastMessageTime = UtilTime::millisecondsCurrent();
	mSocketType = "Server Socket";
	mpContext = pContext;
	mRemoteIp = remoteIp;
	mRemotePort = remotePort;
	mStatus = status;
	mReceiveMessageHeader.create(0, 0);

}

AsioSocketAsync::AsioSocketAsync(
	std::shared_ptr<asio::io_context> pContext,
	ESocketStatus status,
	const std::string remoteIp,
	int remotePort) :
	mSocket(*pContext)
{
	mLastMessageTime = UtilTime::millisecondsCurrent();
	mSocketType = "Client Socket";
	mRemoteIp = remoteIp;
	mRemotePort = remotePort;
	mStatus = status;
	mReceiveMessageHeader.create(0, 0);
}

AsioSocketAsync::~AsioSocketAsync()
{
	// do nothing.
}

void AsioSocketAsync::setEventHandlers(std::shared_ptr<SocketEventHandlerList> pHandlers)
{
	mEventHandlers = pHandlers;
}

void AsioSocketAsync::connect(
	asio::ip::tcp::resolver::results_type& endpoints, 
	std::function<void()> onConnect)
{
	auto self(shared_from_this());

	{
		std::unique_lock<std::recursive_mutex> lock(mStatusMutex);
		if (mStatus != ESocketStatus_Closed)
		{
			return;
		}

		mStatus = ESocketStatus_Connecting;
	}

	asio::async_connect(
		mSocket,
		endpoints,
		[this, self, onConnect](asio::error_code ec, asio::ip::tcp::endpoint)
	{
		if (!ec)
		{
			{
				std::unique_lock<std::recursive_mutex> lock(mStatusMutex);
				mStatus = ESocketStatus_Connected;
			}
			mSocket.set_option(asio::ip::tcp::no_delay(true));

			if (mEventHandlers != nullptr)
			{
				if (!mEventHandlers->onConnected())
				{
					close(nullptr);
					return;
				}
			}

			if (onConnect != nullptr)
			{
				try
				{
					onConnect();
				}
				catch (std::exception& e)
				{
					logError("Exception onConnect: %s", e.what());
					close(nullptr);
				}
				catch (...)
				{
					logError("Exception onConnect Unknown");
					close(nullptr);
				}
			}
		}
		else
		{
			logWarning("%s Connect Failed (%d): %s", mSocketType.c_str(), ec.value(), ec.message().c_str());

			close(nullptr);
		}
	});
}

void AsioSocketAsync::startReceiving()
{
	doReceiveHeader();
}

void AsioSocketAsync::send(const Message& msg)
{
	bool canDoSend = false;

	{
		std::unique_lock<std::recursive_mutex> lock(mSendMessageQueueMutex);
		if (mSendMessageQueue.size() == 0)
		{
			canDoSend = true;
		}
		mSendMessageQueue.push_back(msg);
	}

	if (canDoSend)
	{
		doSend();
	}
}

void AsioSocketAsync::sendHeartbeat()
{
	if (getStatus() != ESocketStatus_Connected)
	{
		return;
	}

	Message msg;
	msg.create(0, 0);
	send(msg);
}

void AsioSocketAsync::doSend()
{
	auto self(shared_from_this());

	const unsigned char* pData = nullptr;
	size_t len = 0;
	{
		std::unique_lock<std::recursive_mutex> lock(mSendMessageQueueMutex);
		if (mSendMessageQueue.size() == 0)
		{
			return;
		}
		const auto& msg = mSendMessageQueue.front();
		pData = msg.getData();
		len = msg.getLength();
	}

	asio::async_write(
		mSocket,
		asio::buffer(pData, len),
		[this, self](asio::error_code ec, std::size_t bytes_transferred)
	{
		if (!ec)
		{
			IoPerf::inst().incr_send_packs(1);
			IoPerf::inst().incr_send_bytes(bytes_transferred);
			bool canDoSend = false;
			{
				std::unique_lock<std::recursive_mutex> lock(mSendMessageQueueMutex);
				mSendMessageQueue.pop_front();
				if (mSendMessageQueue.size() > 0)
				{
					canDoSend = true;
				}
			}
			if (canDoSend)
			{
				doSend();
			}
		}
		else
		{
			logError("%s Write Error (%d): %s", mSocketType.c_str(), ec.value(), ec.message().c_str());

			close(nullptr);
		}
	});
}

void AsioSocketAsync::close(const char* reason)
{
	bool connected = (getStatus() == ESocketStatus_Connected);

	if (!doSetClosed())
	{
		return;
	}

	if (reason != nullptr)
	{
		logInfo("%s Closed. Reason: %s", mSocketType.c_str(), reason);
	}

	try
	{
		if (connected)
		{
			mSocket.shutdown(asio::socket_base::shutdown_both);
		}
		mSocket.close();
	}
	catch (std::exception& e)
	{
		logError("%s Close Error: %s", mSocketType.c_str(), e.what());
	}

	if (mEventHandlers != nullptr)
	{
		mEventHandlers->onDisconnected();
	}
}

void AsioSocketAsync::doReceiveHeader()
{
	auto self(shared_from_this());

	unsigned char* pData = mReceiveMessageHeader.getData();
	size_t len = mReceiveMessageHeader.getLength();

	asio::async_read(mSocket,
		asio::buffer(pData, len), 
		[this, self](asio::error_code ec, std::size_t bytes_transferred)
	{
		if (!ec)
		{
			IoPerf::inst().incr_recv_bytes(bytes_transferred);
			mLastMessageTime = UtilTime::millisecondsCurrent();

			if (mReceiveMessageHeader.parseMsgId() == 0)
			{
				// ignore heart-beat msg
				doReceiveHeader();
				// 每个消息都会有头部，所以在次加一计数即可
				IoPerf::inst().incr_recv_packs(1);
			}
			else if (mReceiveMessageHeader.parseDataBodyLength() > 0)
			{
				doReceiveBody();
			}
			else
			{
				mReceiveMessageProcessed.create(mReceiveMessageHeader.parseMsgId(), 0);

				if (mEventHandlers != nullptr)
				{
					if (!mEventHandlers->onMessage(mReceiveMessageProcessed))
					{
						close(nullptr);
						return;
					}
				}

				doReceiveHeader();
			}
		}
		else
		{
			logWarning("%s Receive Header Failed (%d): %s", mSocketType.c_str(), ec.value(), ec.message().c_str());

			close(nullptr);
		}
	});
}

void AsioSocketAsync::doReceiveBody()
{
	auto self(shared_from_this());

	std::uint32_t dataBodyLength = mReceiveMessageHeader.parseDataBodyLength();

	if (dataBodyLength > Message::MaximalMessageSize)
	{
		logError("%s Message too big: (%d) bytes", mSocketType.c_str(), dataBodyLength);

		close(nullptr);

		return;
	}

	mReceiveMessageProcessed.create(
		mReceiveMessageHeader.parseMsgId(),
		dataBodyLength);

	asio::async_read(mSocket,
		asio::buffer(mReceiveMessageProcessed.getDataBody(), mReceiveMessageProcessed.getDataBodyLength()),
		[this, self](asio::error_code ec, std::size_t bytes_transferred)
	{
		if (!ec)
		{
			IoPerf::inst().incr_recv_bytes(bytes_transferred);
			// handle message
			if (mEventHandlers != nullptr)
			{
				if (!mEventHandlers->onMessage(mReceiveMessageProcessed))
				{
					close(nullptr);
					return;
				}
			}

			doReceiveHeader();
		}
		else
		{
			logError("%s Receive Body Error (%d): %s", mSocketType.c_str(), ec.value(), ec.message().c_str());

			close(nullptr);
		}
	});
}

ESocketStatus AsioSocketAsync::getStatus() const
{
	std::unique_lock<std::recursive_mutex> lock(mStatusMutex);
	return mStatus;
}

const std::string& AsioSocketAsync::getRemoteIp() const
{
	// Never changes, no need to lock.
	return mRemoteIp;
}

int AsioSocketAsync::getRemotePort() const
{
	// Never changes, no need to lock.
	return mRemotePort;
}

unsigned int AsioSocketAsync::getLastMessageTime() const
{
	return mLastMessageTime;
}

bool AsioSocketAsync::doSetClosed()
{
	std::unique_lock<std::recursive_mutex> lock(mStatusMutex);
	if (mStatus == ESocketStatus_Closed)
	{
		return false;
	}
	mStatus = ESocketStatus_Closed;

	return true;
}

CYBERTRON_END
